package com.yykj.message.center.server.service.impl;

import com.yykj.common.util.GsonUtil;
import com.yykj.common.util.LocalStringUtil;
import com.yykj.message.center.client.dto.DrugOrderMessageBodyDTO;
import com.yykj.message.center.client.dto.PrescriptionMessageBodyDTO;
import com.yykj.message.center.client.dto.TDMQMessage;
import com.yykj.message.center.client.dto.TDMQMessageSendResult;
import com.yykj.message.center.client.dto.pay.DrugOrderCancelMessageBodyDTO;
import com.yykj.message.center.client.dto.pay.PayMessageBodyDTO;
import com.yykj.message.center.client.enums.MessageType;
import com.yykj.message.center.client.enums.OrderType;
import com.yykj.message.center.client.enums.PrescriptionAction;
import com.yykj.message.center.server.service.TencentTDMQMessageService;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.*;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;

@Service
@Slf4j
public class TencentTDMQMessageImpl implements TencentTDMQMessageService {

    @Value("${tdmq.serviceUrl}")
    private String serviceUrl;

    @Value("${tdmq.token}")
    private String token;

    @Value("${tdmq.namespace}")
    private String namespace;

    private final String prefix = "persistent://pulsar-pjnbmw52podb/";

    private PulsarClient client;

    @PostConstruct
    public void init() throws Exception {

        client = PulsarClient.builder()
                .serviceUrl(serviceUrl)
                .authentication(AuthenticationFactory.token(token))
                .build();
        System.out.println(">> pulsar client created.");
    }

    @Override
    public TDMQMessageSendResult sendOrderPaySuccessMessage(String orderId, OrderType orderType) throws PulsarClientException {
        String topic = prefix + namespace + "/pay-success";
        TDMQMessage<PayMessageBodyDTO> message = new TDMQMessage<>();
        message.setUuid(LocalStringUtil.getUUID());
        PayMessageBodyDTO bodyDTO = new PayMessageBodyDTO();
        bodyDTO.setOrderId(orderId);
        bodyDTO.setOrderType(orderType);
        message.setData(bodyDTO);
        message.setType(MessageType.PAY_SUCCESS);
        return sendTDMQMessage(topic, message);
    }

    @Override
    public TDMQMessageSendResult sendOrderFinishMessage(String orderId, OrderType orderType) throws PulsarClientException {
        String topic = prefix + namespace + "/task-finish";
        TDMQMessage<PayMessageBodyDTO> message = new TDMQMessage<>();
        message.setUuid(LocalStringUtil.getUUID());
        PayMessageBodyDTO bodyDTO = new PayMessageBodyDTO();
        bodyDTO.setOrderId(orderId);
        bodyDTO.setOrderType(orderType);
        message.setData(bodyDTO);
        message.setType(MessageType.TASK_FINISHED);
        return sendTDMQMessage(topic, message);
    }

    @Override
    public TDMQMessageSendResult sendPrescriptionCreatedMessage(PrescriptionMessageBodyDTO messageBody) throws PulsarClientException {
        String topic = prefix + namespace + "/prescription-created";
        TDMQMessage<PrescriptionMessageBodyDTO> message = new TDMQMessage<>();
        message.setUuid(LocalStringUtil.getUUID());

        messageBody.setAction(PrescriptionAction.create);
        messageBody.setStatus(1);
        message.setData(messageBody);
        message.setType(MessageType.PRESCRIPTION_CREATED);
        return sendTDMQMessage(topic, message);
    }

    @Override
    public TDMQMessageSendResult sendPrescriptionChangeMessage(PrescriptionMessageBodyDTO messageBody) throws PulsarClientException  {
        String topic = prefix + namespace + "/prescription-change";
        TDMQMessage<PrescriptionMessageBodyDTO> message = new TDMQMessage<>();
        message.setUuid(LocalStringUtil.getUUID());
        messageBody.setAction(PrescriptionAction.change);
        message.setData(messageBody);
        message.setType(MessageType.PRESCRIPTION_STATUS_CHANGE);
        return sendTDMQMessage(topic, message);
    }

    @Override
    public TDMQMessageSendResult sendDrugOrderMessage(DrugOrderMessageBodyDTO messageBody) throws PulsarClientException {
        String topic = prefix + namespace + "/prescription-drug-order";
        TDMQMessage<DrugOrderMessageBodyDTO> message = new TDMQMessage<>();
        message.setUuid(LocalStringUtil.getUUID());
        message.setData(messageBody);
        message.setType(MessageType.DRUG_ORDER_CREATE);
        return sendTDMQMessage(topic, message);
    }

    @Override
    public TDMQMessageSendResult sendCancelDrugOrderMessage(DrugOrderCancelMessageBodyDTO messageBody) throws PulsarClientException {

        String topic = prefix + namespace + "/drug-cancel-order";
        TDMQMessage<DrugOrderCancelMessageBodyDTO> message = new TDMQMessage<>();
        message.setUuid(LocalStringUtil.getUUID());
        message.setData(messageBody);
        message.setType(MessageType.DRUG_ORDER_CANCEL);
        return sendDelayTDMQMessage(topic, message,messageBody.getCancelTime());
    }


    private TDMQMessageSendResult sendTDMQMessage(String topic , TDMQMessage message) throws PulsarClientException {
        //创建生产者
        Producer<byte[]> producer = client.newProducer()
                .topic(topic)
                .create();
        System.out.println(">> pulsar producer created.");

        //同步发送消息
        MessageId msgId = producer.newMessage().value(GsonUtil.toJson(message).getBytes()).send();
        log.info("Message = {}", GsonUtil.toJson(message));
        TDMQMessageSendResult result = new TDMQMessageSendResult();
        result.setMsgId(msgId.toString());
        result.setUuid(message.getUuid());
        //关闭生产者
        producer.close();

        return result;
    }

    private TDMQMessageSendResult sendDelayTDMQMessage(String topic , TDMQMessage message,Long cancelTime) throws PulsarClientException {
        //创建生产者
        Producer<byte[]> producer = client.newProducer()
                .topic(topic)
                .create();
        System.out.println(">> pulsar producer created.");
        TDMQMessageSendResult result = new TDMQMessageSendResult();
        try{
            MessageId msgId = producer.newMessage().value(GsonUtil.toJson(message).getBytes()).deliverAt(cancelTime).send();
            log.info("Message = {}", GsonUtil.toJson(message));
            result.setMsgId(msgId.toString());
            result.setUuid(message.getUuid());
        }catch (Exception e) {
            e.printStackTrace();
        }
        //关闭生产者
        producer.close();
        return result;
    }
}
